﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace ProducerTest1
{
    public class SinglePush : IPush
    {
        public void Invoke(ProducerConfig config)
        {
            throw new NotImplementedException();
        }

        public async Task InvokeAsync(ProducerConfig config)
        {
            using (var producer = new ProducerBuilder<string, string>(config).Build())
            {
                try
                {
                    //不要看着这里使用的是异步方法，其实这种方式对于kafka的连接来说叫做“同步推送”
                    DeliveryResult<string, string> result = await producer.ProduceAsync("CustomerCountry", new Message<string, string> { Key = "Precision Products", Value = "France" });
                    Console.WriteLine($"delivered to: {result.TopicPartitionOffset}");
                }
                catch (ProduceException<string, string> e)
                {
                    Console.WriteLine($"failed to deliver message:{e.Message} [{e.Error.Code}]");
                }
            }
        }
    }
}
